package com.atguigu.flink.sqlfunction;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2022/12/23
 */
public class Demo1_FunctionUse
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String  createTableSQL = " CREATE TABLE t1( id string, ts bigint , vc int  ) " +
            "                       WITH (  " +
            "                         'connector' = 'filesystem',  " +
            "                         'path' = 'data/t1',   " +
            "                         'format' = 'csv'    " +
            "                            )      ";


        //建表(连接外部文件系统)
        tableEnv.executeSql(createTableSQL);

        Table t1 = tableEnv.from("t1");

        //使用函数  官方提供 https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/functions/systemfunctions/

        //DSL(tableapi)中如何使用  调用随机函数
        /*t1.select($("id"),rand($("vc")).as("rum"))
          .execute()
          .print();*/

        //SQL
        tableEnv.sqlQuery("select id , rand(vc) rum from t1")
                .execute()
                .print();


    }
}
